package rxjava;

import android.app.Activity;
import android.graphics.Bitmap;
import android.os.Bundle;
import android.util.Log;
import java.io.IOException;
import java.io.Serializable;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.List;
import org.jaaksi.commondemo.model.Person;
import rx.Observable;
import rx.Observer;
import rx.Single;
import rx.SingleSubscriber;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Action2;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

/**
 * Observable<Object> create(OnSubscribe<Object>)
 * Observable内部持有OnSubscribe<T> onSubscribe
 * interface OnSubscribe<T> extends Action1<Subscriber<? super T>>
 * subscribeOn()是在call()方法中起作用，而observeOn()是在onNext()中作用。
 *
 * > * 下面提到的“操作”包括产生事件、用操作符操作事件以及最终的通过 subscriber 消费事件；
 * > * 只有第一subscribeOn() 起作用（所以多个 subscribeOn() 无意义）；
 * > * 这个 subscribeOn() 控制从流程开始的第一个操作，直到遇到第一个 observeOn()；
 * > * observeOn() 可以使用多次，每个 observeOn() 将导致一次线程切换()，这次切换开始于这次 observeOn() 的下一个操作；
 * > * 不论是 subscribeOn() 还是 observeOn()，每次线程切换如果不受到下一个 observeOn() 的干预，线程将不再改变，不会自动切换到其他线程。
 */
public class TestRxJavaActivity extends Activity {
  private final String TAG = "TestRxJavaActivity";

  @Override protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    //changeThread();
    //login();
    //zip();
    //merge();
  }

  /**
   * 操作符返回一个Obversable，它使用这个函数按顺序结合两个或多个Observables发射的数据项，然后它发射这个函数返回的结果。
   * 如果任何一个返回了onError或者没有调用onNext或onComplete就不会回调zip 的最后一个参数的call方法
   * 它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据
   */
  private void zip() {
    // observable1在执行的过程中休眠了5s,而5s后执行了observable 1 weak之后才执行observable2 observable 2..
    // 且必须每个都发射（onNext）了数据zip才会发射数据，否则zip不会发射数据，而且subscribe不会回调onNext方法
    // 只要有任何一个掉了onError，subscribe就会回调onError方法，如果有任何一个没有掉任何方法，就不会回调subscribe
    Observable<String> observable1 = Observable.create(new Observable.OnSubscribe<String>() {
      @Override public void call(Subscriber<? super String> subscriber) {
        Log.i(TAG, "observable 1 sleep");
        System.out.println("observable1 thread = " + Thread.currentThread().getName());
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
        subscriber.onNext("observable1 on next");
        Log.i(TAG, "observable 1 weak");
        //subscriber.onCompleted();
        subscriber.onError(new Throwable("每个都发射（onNext）了数据zip才会发射数据"));
      }
    }).subscribeOn(AndroidSchedulers.mainThread());  // observable执行在io线程中

    Observable<Integer> observable2 = Observable.create(new Observable.OnSubscribe<Integer>() {
      @Override public void call(Subscriber<? super Integer> subscriber) {
        Log.i(TAG, "observable 2..");
        System.out.println("observable2 thread = " + Thread.currentThread().getName());
        subscriber.onNext(2);
        Log.i(TAG, "observable 2 onNext");
        subscriber.onCompleted();
      }
    }).subscribeOn(Schedulers.io());

    Observable.zip(observable1, observable2, new Func2<String, Integer, Boolean>() {
      @Override public Boolean call(String s, Integer integer) {
        Log.i(TAG, "zip... " + integer + "," + s);
        return true;
      }
    })
      //.subscribeOn(AndroidSchedulers.mainThread())
      //.subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<Boolean>() {
      @Override public void onCompleted() {
        Log.i(TAG, "onCompleted...");
      }

      @Override public void onError(Throwable e) {
        Log.i(TAG, "onError...");
        System.out.println("subscribe thread = " + Thread.currentThread().getName());
      }

      @Override public void onNext(Boolean aBoolean) {
        Log.i(TAG, "onNext...");
      }
    });
  }

  /*public Subscription subscribe(Subscriber subscriber) {
    // 伪代码：subscribe方法
    subscriber.onStart();
    // 当 Observable 被订阅的时候，就会回调OnSubscribe 的 call() 方法
    onSubscribe.call(subscriber);
    return subscriber;
  }*/

  /*private void action(){
    // 解释
    // 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
    observable.subscribe(onNextAction);
    // 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
    observable.subscribe(onNextAction, onErrorAction);
    // 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
    observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
  }*/

  /**
   * subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
   * observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程
   * 通过 observeOn() 的多次调用，程序实现了线程的多次切换
   * 不同于 observeOn() ， subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的
   */
  private void changeThread() {
    // just发生在 main 线程，由 subscribeOn() 指定，即事件产生的线程
    Observable.just(1, 2, 3, 4, 5, 6)
      .subscribeOn(AndroidSchedulers.mainThread())
      .observeOn(Schedulers.newThread())
      .filter(new Func1<Integer, Boolean>() { // 过滤掉奇数，filter 发生在newThread,由observeOn指定，消费事件
        @Override public Boolean call(Integer integer) {
          return integer % 2 == 0;
        }
      })
      .observeOn(Schedulers.io())
      .map(new Func1<Integer, String>() { // 转换数据，发生在io线程，有observeOn指定
        @Override public String call(Integer integer) {
          return String.valueOf(integer);
        }
      })
      .observeOn(AndroidSchedulers.mainThread()) // Android 主线程，由 observeOn() 指定
      .subscribe(new Action1<String>() {
        @Override public void call(String s) {
        }
      });
  }

  /**
   * merge只是合并不会去重，且不保证顺序，如果有的慢，发射的也会慢，如果要保证顺序使用contact
   */
  private void merge() {
    Observable<Integer> observable1 = Observable.just(1, 2, 3);
    Observable<Integer> observable2 = Observable.just(2, 3, 4);
    Observable<String> observable3 = Observable.just("3", "9", "9");

    Observable.merge(observable1, observable2).subscribe(new Action1<Integer>() {
      @Override public void call(Integer integer) {
        // 1 2 3 2 3 4
        Log.e("merge", String.valueOf(integer));
      }
    });

    // 如果类型不同，就会找公共父类
    Observable.merge(observable1, observable3).subscribe(new Action1<Serializable>() {
      @Override public void call(Serializable serializable) {

      }
    });
  }

  /**
   * 不回调onNext，只回调onCompleted
   */
  private void empty() {
    Observable.empty().subscribe(new Subscriber<Object>() {
      @Override public void onCompleted() {

      }

      @Override public void onError(Throwable e) {

      }

      @Override public void onNext(Object o) {

      }
    });
  }

  /**
   * retryWhen 根据onError中抛出的异常类型处理自己想要重试的情况
   */
  private void retryWhen() {
    Observable.just(1).retry(new Func2<Integer, Throwable, Boolean>() { // 这样也可以？？
      @Override public Boolean call(Integer integer, Throwable throwable) {
        if (throwable instanceof SocketTimeoutException && integer < 2) {
          return true;
        } else {
          return false;
        }
      }
    });

    Observable.just(1)
      .retry(3)
      .retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
        @Override public Observable<?> call(final Observable<? extends Throwable> observable) {
          return observable.flatMap(new Func1<Throwable, Observable<?>>() {
            @Override public Observable<?> call(Throwable throwable) {
              // 通过flatMap拿到拿到 Throwable以及返回Observable,判断异常类型分情况处理
              if (throwable instanceof IOException) {
                // 如果是IO异常重试否则不重试
                return Observable.just(null); // ？？？
              }
              // 否则不重试，抛出异常错误
              return Observable.error(throwable);
            }
          });
        }
      })
      .subscribe(new Subscriber<Integer>() {
        @Override public void onCompleted() {

        }

        @Override public void onError(Throwable e) {

        }

        @Override public void onNext(Integer integer) {

        }
      });
  }

  private void from() {
    Integer[] arrs = { 1, 2, 3 };
    // 除了 subscribe(Observer) 和 subscribe(Subscriber) ，
    // subscribe() 还支持不完整定义的回调，RxJava 会自动根据定义创建出 Subscriber 。
    Observable.from(arrs).subscribe(new Action1<Integer>() {
      @Override public void call(Integer integer) {
        Log.i(TAG, "call:" + integer);
      }
    });
  }

  private void fun1() {
    Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
      @Override public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; i < 3; i++) {
          subscriber.onNext(i);
        }
        // onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。
        // RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
        subscriber.onCompleted();
      }
    });

    // Observable 通过subscribe方法和Subscriber实现了订阅关系
    Subscription subscription = observable.subscribe(new Subscriber<Integer>() {
      @Override public void onCompleted() {
        Log.i(TAG, "onCompleted");
      }

      @Override public void onError(Throwable e) {

      }

      @Override public void onNext(Integer integer) {
        Log.i(TAG, "onNext:" + integer);
      }
    });

    /*在 RxJava 的默认规则中，事件的发出和消费都是在同一个线程的。也就是说，如果只用上面的方法，实现出来的只是一个同步的观察者模式。
    观察者模式本身的目的就是『后台处理，前台回调』的异步机制，
    因此异步对于 RxJava 是至关重要的。而要实现异步，则需要用到 RxJava 的另一个概念：Scheduler*/

    /*unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。
    在这个方法被调用后，Subscriber 将不再接收事件。
    一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。
    unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，
    这个引用如果不能及时被释放，将有内存泄露的风险。
    所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）
    调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。*/
    // 伪代码
    if (subscription.isUnsubscribed()) {
      subscription.unsubscribe();
    }
  }

  private void schedulers() {
    Integer[] arrs = { 1, 2, 3 };
    // create() 方法是 RxJava 最基本的创造事件序列的方法。基于这个方法， RxJava 还提供了一些方法用来快捷创建事件队列
    // just(T...): 将传入的参数依次发送出来
    // 将会依次调用：
    // onNext("Hello");
    // onNext("Hi");
    // onNext("Aloha");
    // onCompleted();
    Observable.just(1, 2, 3, 4).subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
      .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
      .subscribe(new Action1<Integer>() {
        @Override public void call(Integer integer) {

        }
      });
  }

  private void map() {
    Observable.just("imges/icon.png") // 输入类型 String
      .map(new Func1<String, Bitmap>() { // 转化为 Bitmap
        @Override public Bitmap call(String s) {
          //return getBitmapFromPath(filePath); // 返回类型 Bitmap
          return null;
        }
      }).subscribe(new Action1<Bitmap>() {
      @Override public void call(Bitmap bitmap) {
        // 加载bitmap
        //showBitmap(bitmap);
      }
    });
  }

  /**
   * Single类似于Observable，不同的是，它总是只发射一个值，或者一个错误通知，而不是发射一系列的值。
   * 因此，不同于Observable需要三个方法onNext, onError, onCompleted，订阅Single只需要两个方法：
   * onSuccess - Single发射单个的值到这个方法
   * onError - 如果无法发射需要的值，Single发射一个Throwable对象到这个方法
   * Single只会调用这两个方法中的一个，而且只会调用一次，调用了任何一个方法之后，订阅关系终止。
   */
  private void single() {
    Single.create(new Single.OnSubscribe<Integer>() {
      @Override public void call(SingleSubscriber<? super Integer> singleSubscriber) {

      }
    }).compose(new Single.Transformer<Integer, String>() {
      @Override public Single<String> call(Single<Integer> integerSingle) {
        return null;
      }
    }).subscribe(new SingleSubscriber<String>() {
      @Override public void onSuccess(String value) {

      }

      @Override public void onError(Throwable error) {

      }
    });
  }

  //private void flatMap() {
  //  Student[] students =
  //    DataParseUtil.mGson.fromJson(RawUtils.getString(this, R.raw.student), Student[].class);
  //
  //  Observable.from(students).flatMap(new Func1<Student, Observable<Student.Course>>() {
  //    @Override public Observable<Student.Course> call(Student student) {
  //      return Observable.from(student.courseList);
  //    }
  //  }).subscribe(new Action1<Student.Course>() {
  //    @Override public void call(Student.Course course) {
  //      LogUtil.i(TAG, course.name);
  //    }
  //  });
  //}

  //private void lift(){
  //  Student[] students =
  //      DataParseUtil.mGson.fromJson(RawUtils.getString(this, R.raw.student), Student[].class);
  //  Observable.from(students).compose()
  //}

  public void collect() {
    // 从List<FrameInfo> 中筛选出frameInfo的id不为null的，并且去除images放到一个新的List集合中
    List<Person> frameInfoList = new ArrayList<>();
    Observable.from(frameInfoList).filter(new Func1<Person, Boolean>() {
      @Override public Boolean call(Person frameInfo) {
        return frameInfo.id != null;
      }
    }).flatMap(new Func1<Person, Observable<String>>() {
      @Override public Observable<String> call(Person frameInfo) {
        return Observable.just(frameInfo.name);
      }
    }).collect(new Func0<List<String>>() {
      @Override public List<String> call() {
        return new ArrayList<>();
      }
    }, new Action2<List<String>, String>() {
      @Override public void call(List<String> list, String s) {
        list.add(s);
      }
    }).subscribe(new Action1<List<String>>() {
      @Override public void call(List<String> list) {
        // TODO

      }
    });
  }
}
